/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package br.ufba.lasid.ds.fd;

import br.ufba.lasid.base.api.IResource;
import br.ufba.lasid.comm.HostList;
import br.ufba.lasid.comm.api.ICommunicator;
import br.ufba.lasid.comm.api.IHost;
import br.ufba.lasid.ds.fd.api.IFailureDetector;
import br.ufba.lasid.ds.fd.api.IFailureDetector.ModuleTypes;
import br.ufba.lasid.ds.fd.api.IFailureDetectorPeriodicActionModule;
import br.ufba.lasid.utils.Agenda;
import br.ufba.lasid.utils.Schedule;
import java.net.ConnectException;
import java.net.NoRouteToHostException;
import java.util.Calendar;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 *
 * @author aliriosa
 */
public class FailureDetectorPeriodicAction implements IFailureDetectorPeriodicActionModule {

    private IFailureDetector detector;
    private long period = 100L;
    private static final Log logger = LogFactory.getLog(FailureDetector.class);
    private ICommunicator communicator;
    private IFailureDetector.MonitoringStyles style;

    public FailureDetectorPeriodicAction(IFailureDetector detector) {
        this.detector = detector;
    }

    public FailureDetectorPeriodicAction() {
    }

    @Override
    public String getName() throws Exception {
        return this.getModuleType().name();
    }

    @Override
    public ModuleTypes getModuleType() throws Exception {
        return ModuleTypes.PeriodicAction;
    }

    @Override
    public void setFailureDetector(IFailureDetector detector) throws Exception {
        this.detector = detector;
    }

    @Override
    public synchronized void setPeriod(long period) {
        this.period = period;
    }

    @Override
    public synchronized long getPeriod() {
        return this.period;
    }

    @Override
    public void startup() throws Exception {
        new Thread(this).start();
    }

    @Override
    public void run() {
        try {
            communicator = detector.getCommunicator();
            style = detector.getMonitoringStyle();

            while (true) {
                try {
                    perform();
                    Thread.sleep(getPeriod());
                } catch (Exception ex) {
                    logger.error("A runtime error has been detected.", ex);
                }
            }
        } catch (Exception ex) {
            logger.error("couldn't start", ex);
        }
    }

    @Override
    public void perform() throws Exception {

        HostList hosts = new HostList();
        hosts.addAll(detector.getRegisteredHosts());

        /**
         * If the failure detector uses a pull monitoring style, then it'll
         * periodically send a check state message for all other registered
         * hosts.
         */
        if (style.equals(IFailureDetector.MonitoringStyles.Pull)) {
            /**
             * Build a message which will ask the state of the resources in the
             * others hosts.
             */
            String message = IFailureDetector.Utils.newMessage(
                    IFailureDetector.MessageKinds.CheckState);

            for (IHost host : hosts) {
                if (!host.isLocalhost()) {
//                    String hostID = host.getID();                    
//                    try{
//                        communicator.send(hostID, detector.getName(), message);  
//                    }catch(ConnectException ce){
//                        host.setState(IResource.States.Down);                        
//                        //ce.printStackTrace();
//                    }catch(NoRouteToHostException nre){
//                        host.setState(IResource.States.Down);
//                    }catch(Exception e){
//                        e.printStackTrace();
//                    }

                    /**
                     * Instantiate a new thread at each sending instant can
                     * bring some performance issues -- we must review it later.
                     *
                     *
                     */
                    new Sender(host, message).start();

                    if (!host.isDown()) {
                        /**
                         * I think it isn't right here -- we must revaluate who
                         * is responsible of scheduling and canceling timeouts
                         * -- maybe the failure detector class, for instance: -
                         * FD.scheduleTimeout(host) | FD.cancelTimeout(host).
                         */
                        scheduleTimeout(host);
                    } else {
                        cancelTimeout(host);
                    }
                }
            }
        } else {
            /**
             * If the failure detector uses a push monitoring style, then it'll
             * periodically reports its state to the other registered hosts.
             */
            if (style.equals(IFailureDetector.MonitoringStyles.Push)) {
                /**
                 * When o failure detector's state reporter is started up, it
                 * must send the local resource states to the other hosts.
                 */
                detector.getStateReporter().startup();
            }
        }
    }

    public void scheduleTimeout(IHost host) throws Exception {
        String hostID = host.getID();
        final Agenda agenda = detector.getAgenda();

        long now = Calendar.getInstance().getTimeInMillis();
        long timeout = now + detector.getTimeout();

        synchronized (agenda) {
            if (!agenda.containsKey(hostID)) {
                agenda.put(hostID, new Schedule());
                agenda.notify();
            }
        }

        final Schedule schedule = agenda.get(hostID);

        synchronized (schedule) {
            if (!schedule.containsKey(timeout)) {
                schedule.put(timeout, host);
                schedule.notify();
            }
        }
    }

    public void cancelTimeout(IHost host) throws Exception {
        String hostID = host.getID();
        final Agenda agenda = detector.getAgenda();

        synchronized (agenda) {
            if (!agenda.containsKey(hostID)) {
                agenda.put(hostID, new Schedule());
                agenda.notify();
            }
        }

        final Schedule schedule = agenda.get(hostID);

        synchronized (schedule) {
            schedule.clear();
            schedule.notify();
        }
    }

    class Sender extends Thread {

        IHost host;
        String message;

        public Sender() {
        }

        public Sender(IHost host, String message) {
            this.host = host;
            this.message = message;
        }

        @Override
        public void run() {
            String hostID = host.getID();
            try {
                communicator.send(hostID, detector.getName(), message);
            } catch (ConnectException ce) {
                host.setState(IResource.States.Down);
                //ce.printStackTrace();
            } catch (NoRouteToHostException nre) {
                host.setState(IResource.States.Down);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
